package cn.redragon.soa.common.config.rabbit;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.rabbitmq.client.Channel;
import java.time.Duration;
import lombok.extern.slf4j.Slf4j;
import org.aopalliance.aop.Advice;
import org.springframework.amqp.AmqpRejectAndDontRequeueException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.annotation.EnableRabbit;
import org.springframework.amqp.rabbit.annotation.RabbitListenerConfigurer;
import org.springframework.amqp.rabbit.config.RetryInterceptorBuilder;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.RabbitListenerEndpointRegistrar;
import org.springframework.amqp.rabbit.retry.MessageRecoverer;
import org.springframework.amqp.rabbit.retry.RejectAndDontRequeueRecoverer;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.context.annotation.Bean;
import org.springframework.messaging.handler.annotation.support.DefaultMessageHandlerMethodFactory;
import org.springframework.retry.RetryPolicy;
import org.springframework.retry.backoff.ExponentialBackOffPolicy;
import org.springframework.retry.interceptor.RetryOperationsInterceptor;
import org.springframework.retry.policy.NeverRetryPolicy;
import org.springframework.retry.policy.SimpleRetryPolicy;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.validation.beanvalidation.LocalValidatorFactoryBean;

@EnableRabbit
@ConditionalOnClass({RabbitTemplate.class, Channel.class})
@Slf4j
public class CommonRabbitConfig implements RabbitListenerConfigurer {

    @Value("${spring.rabbitmq.template.reply-timeout:45s}")
    private Duration replyTimeout;

    @Qualifier("commonObjectMapper")
    @Autowired
    private ObjectMapper objectMapper;

    @Bean
    public SimpleRabbitListenerContainerFactory containerFactory(ConnectionFactory connectionFactory, final Advice retryAdvice) {
        SimpleRabbitListenerContainerFactory containerFactory = new SimpleRabbitListenerContainerFactory();
        containerFactory.setConnectionFactory(connectionFactory);
        containerFactory.setAdviceChain(retryAdvice);
        containerFactory.setMessageConverter(messageConverter());
        containerFactory.setDefaultRequeueRejected(Boolean.FALSE);
        //Recover mode # https://docs.spring.io/spring-amqp/reference/html/
        containerFactory.setMissingQueuesFatal(false);
        MessagePostProcessor[] postProcessors = getMessagePostProcessors();
        if (postProcessors != null) {
            containerFactory.setAfterReceivePostProcessors(postProcessors);
        }

        return containerFactory;
    }

    @Bean
    public RetryOperationsInterceptor retryAdvice(final MessageRecoverer messageRecoverer, final RetryProperties retryConfig) {
        return RetryInterceptorBuilder.stateless().retryOperations(retryTemplate(retryConfig)).recoverer(messageRecoverer).build();
    }

    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate template = new RabbitTemplate(connectionFactory);
        template.setMessageConverter(messageConverter());
        template.setReplyTimeout(replyTimeout.toMillis());
        return template;
    }

    @Bean
    public MessageRecoverer messageRecoverer() {
        return new RejectAndDontRequeueRecoverer() {
            @Override
            public void recover(Message message, Throwable cause) {
                try {
                    log.warn("Recovering message {}", message);
                    log.warn(message.toString(), cause);
                } catch (Exception e) { // Catch all exceptions here to ensure the consumer won't restart indefinitely
                    log.error("Unexpected error in MessageRecoverer", e);
                }
                throw new AmqpRejectAndDontRequeueException(cause);
            }
        };
    }

    @Bean
    public Jackson2JsonMessageConverter messageConverter() {
        return new Jackson2JsonMessageConverter(objectMapper);
    }

    @Override
    public void configureRabbitListeners(RabbitListenerEndpointRegistrar registrar) {
        registrar.setMessageHandlerMethodFactory(validatingHandlerMethodFactory());
    }

    private RetryTemplate retryTemplate(final RetryProperties retryProperties) {
        final RetryTemplate retryTemplate = new RetryTemplate();

        final RetryPolicy retryPolicy;
        if (retryProperties.isEnabled()) {
            retryPolicy = new SimpleRetryPolicy(retryProperties.getMaxAttempts(), retryProperties.getRetryableExceptions(), true);
            final ExponentialBackOffPolicy backOffPolicy = new ExponentialBackOffPolicy();
            backOffPolicy.setInitialInterval(retryProperties.getInitialInterval());
            backOffPolicy.setMultiplier(retryProperties.getMultiplier());
            backOffPolicy.setMaxInterval(retryProperties.getMaxInterval());
            retryTemplate.setBackOffPolicy(backOffPolicy);
        } else {
            retryPolicy = new NeverRetryPolicy();
        }
        retryTemplate.setRetryPolicy(retryPolicy);
        return retryTemplate;
    }

    // Message Validation
    private DefaultMessageHandlerMethodFactory validatingHandlerMethodFactory() {
        DefaultMessageHandlerMethodFactory factory = new DefaultMessageHandlerMethodFactory();
        factory.setValidator(localValidatorFactoryBean());
        factory.afterPropertiesSet();
        return factory;
    }

    private LocalValidatorFactoryBean localValidatorFactoryBean() {
        LocalValidatorFactoryBean validator = new LocalValidatorFactoryBean();
        validator.afterPropertiesSet();
        return validator;
    }

    protected MessagePostProcessor[] getMessagePostProcessors() {
        return null;
    }
}
